"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 故障&可靠性测试
Case Name   : 延迟等待过程中删除待同步对象（表、函数，触发器）
Create At   : 2021/03/09
@zou_jialiang0505328126
Description :
    1.设置延迟为5min
    2.切换备节点为级联备
    3.创建表，函数，触发器，序列
    4.等待30s，删除表、函数及触发器、序列对象
    5.查询延迟等待时间内数据库对象变化
Expect      :
    1.设置成功
    2.切换成功
    3.建表成功
    4.删除成功
    5.查询延迟等待时间内数据库对象变化（无--》有--》无）
History     :
    Modified by @wan005 2021/7/14: 调整wal_sender_timeout,规避因环境性能差导致的失败,取消赋权
    Modified by @wan005 2021/7/22: 打开most_available_sync，避免因为备机异常导致的主机查询超时
    Modified by @wan005 2021/8/4 : 先drop再设置备机延迟 校验结果
    Modified by @peilinqian 2021/10/21: 备份文件恢复数据库存在备机状态异常、启动超时、
    主备连接不通的问题，优化用例逻辑，删除备份恢复操作，直接验证（表、函数、触发器）延迟备份功能
    Modified by @wan005 2021/11/05: 延长recovery_min_apply_delay为5min
"""

import time
import unittest

from testcase.utils.CommonSH import CommonSH
from testcase.utils.CommonSH import RestartDbCluster
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node


class RecoveryDelay(unittest.TestCase):
    nodes_tuple = ('PrimaryDbUser', 'Standby1DbUser', 'Standby2DbUser')

    @RestartDbCluster(*nodes_tuple)
    def setUp(self):
        self.log = Logger()
        self.log.info('----Opengauss_Reliability_Dbsys_Case262.py 开始执行----')
        self.pri_root = Node(node='PrimaryRoot')
        self.sta_root = [Node(node='Standby1Root'),
                         Node(node='Standby2Root')]
        self.sta_sh = [CommonSH('Standby1DbUser'),
                       CommonSH('Standby2DbUser')]
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.db_name = 'db_dbsys_case262'
        self.tb_name = 'tb_dbsys_case262'
        self.trigger_name = 'trigger_dbsys_case262'
        self.func_name = 'func_dbsys_case262'
        self.constant = Constant()

        self.log.info('====安装ntp服务====')
        install_cmd = "yum install -y ntp && " \
                      "systemctl enable ntpd && systemctl start ntpd"
        for i in range(2):
            result = self.sta_root[i].sh(install_cmd).result()
            self.log.info(result)

        self.log.info('----同步集群时间----')
        for i in range(2):
            current = self.pri_root.sh(
                "date \"+%m/%d/%Y %H:%M:%S\"").result()
            self.log.info(current)
            datecmd = f'date -s "{current}";hwclock --systohc;' \
                f'service ntpd stop;ntpdate ntp.api.bz;service ntpd start'
            self.log.info(datecmd)
            result = self.sta_root[i].sh(datecmd).result()
            self.log.info(result)

        self.log.info('----查询相关参数初始配置值----')
        result = self.pri_sh.executDbSql('show wal_sender_timeout;')
        self.log.info(f"wal_sender_timeout is {result}")
        self.para1 = result.strip().splitlines()[-2]
        result = self.pri_sh.executDbSql('show most_available_sync;')
        self.log.info(f"most_available_sync is {result}")
        self.para2 = result.strip().splitlines()[-2]
        result = self.pri_sh.executDbSql('show recovery_max_workers;')
        self.log.info(f"recovery_max_workers is {result}")
        self.para3 = result.strip().splitlines()[-2]

        self.log.info("----创建数据库----")
        create_db_cmd = f'drop database if exists {self.db_name};' \
            f'create database {self.db_name};'
        create_db_result = self.pri_sh.executDbSql(create_db_cmd)
        self.log.info(create_db_result)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS_MSG,
                      create_db_result)

    def test_recovery_delay(self):
        step_txt = '----step1:设置延迟为5min; expect:设置成功----'
        self.log.info(step_txt)
        result = self.pri_sh.executeGsguc('reload',
                                          self.constant.GSGUC_SUCCESS_MSG,
                                          "recovery_min_apply_delay='5min'")
        self.assertTrue(result, '执行失败:' + step_txt)

        step_txt = '----step2:切换备节点为级联备; expect:切换成功----'
        self.log.info(step_txt)
        result = self.sta_sh[0].buildStandby('-M cascade_standby')
        self.assertIn(self.constant.REBUILD_SUCCESS_MSG, result,
                      '执行失败:' + step_txt)
        result = self.sta_sh[0].exec_refresh_conf()
        self.assertTrue(result, '执行失败:' + step_txt)
        self.log.info('----等待主备一致----')
        result = self.sta_sh[1].check_data_consistency()
        self.assertTrue(result, '执行失败:' + step_txt)
        for i in range(90):
            result = self.pri_sh.check_cascade_standby_consistency()
            if result:
                break
            time.sleep(20)
        self.assertTrue(result, '执行失败:' + step_txt)
        self.log.info('----查询主备状态----')
        self.pri_sh.getDbClusterStatus('status')

        step_txt = '----step3:创建表，函数，触发器，序列; expect:创建成功----'
        self.log.info(step_txt)
        self.log.info('----创建表----')
        sql = f"create table {self.tb_name}(num int)" \
            f"partition by range(num)" \
            f"(partition num1 values less than(10000), " \
            f"partition num2 values less than(20000)," \
            f"partition num3 values less than(300000));"
        result = self.pri_sh.executDbSql(sql, self.db_name)
        self.log.info(result)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, result,
                      '执行失败:' + step_txt)
        self.log.info('----创建函数 ----')
        sql = f"create or replace function {self.func_name}() " \
            f"returns trigger as  " \
            f"\$\$   " \
            f"declare   " \
            f"begin  " \
            f"insert into {self.tb_name} " \
            f"values(new.id1);  " \
            f"return new;      " \
            f"end           " \
            f"\$\$ language plpgsql;"
        result = self.pri_sh.executDbSql(sql, self.db_name)
        self.log.info(result)
        self.assertIn('CREATE FUNCTION', result, '执行失败:' + step_txt)
        self.log.info('----创建触发器 ----')
        sql = f" create trigger {self.trigger_name} " \
            f"  before insert on {self.tb_name}" \
            f"  for each row " \
            f" execute procedure {self.func_name}();"
        result = self.pri_sh.executDbSql(sql, self.db_name)
        self.log.info(result)
        self.assertIn('CREATE TRIGGER', result, '执行失败:' + step_txt)
        self.log.info('----创建SEQUENCE ----')
        sql = f"create sequence serial start 101 cache 20;"
        result = self.pri_sh.executDbSql(sql, self.db_name)
        self.log.info(result)
        self.assertIn('CREATE SEQUENCE', result, '执行失败:' + step_txt)

        step_txt = '----step4:等待30s，删除表、函数及触发器、序列对象; expect:删除成功----'
        self.log.info(step_txt)
        self.log.info('----等待30s----')
        time.sleep(60*2)
        self.log.info('----删除数据库对象----')
        sql = f"drop function if exists {self.func_name}() cascade;"
        result = self.pri_sh.executDbSql(sql, self.db_name)
        self.log.info(result)
        self.assertIn('DROP FUNCTION', result, '执行失败:' + step_txt)
        sql = f"drop table if exists {self.tb_name};"
        result = self.pri_sh.executDbSql(sql, self.db_name)
        self.log.info(result)
        self.assertIn('DROP TABLE', result, '执行失败:' + step_txt)
        sql = f"drop sequence if exists serial;"
        result = self.pri_sh.executDbSql(sql, self.db_name)
        self.log.info(result)
        self.assertIn('DROP SEQUENCE', result, '执行失败:' + step_txt)

        step_txt = '----step5:查询延迟等待时间内数据库对象变化; expect:无--》有--》无----'
        self.log.info(step_txt)
        self.log.info('----创建操作未同步，查询各节点数据库对象，备机无----')
        sql = f"select sysdate;select tgname from pg_trigger;"
        result = self.pri_sh.executDbSql(sql, self.db_name)
        self.log.info(result)
        self.assertNotIn(self.trigger_name, result, '执行失败:' + step_txt)
        for i in range(2):
            result = self.sta_sh[i].executDbSql(sql, self.db_name)
            self.log.info(result)
            self.assertNotIn(self.trigger_name, result, '执行失败:' + step_txt)
        sql = f"select sysdate;\sf {self.func_name}"
        result = self.pri_sh.executDbSql(sql, self.db_name)
        self.log.info(result)
        self.assertIn(self.constant.NOT_EXIST, result, '执行失败:' + step_txt)
        for i in range(2):
            result = self.sta_sh[i].executDbSql(sql, self.db_name)
            self.log.info(result)
            self.assertIn(self.constant.NOT_EXIST, result, '执行失败:' + step_txt)
        sql = f"select sysdate;\ds"
        result = self.pri_sh.executDbSql(sql, self.db_name)
        self.log.info(result)
        self.assertNotIn('serial', result, '执行失败:' + step_txt)
        for i in range(2):
            result = self.sta_sh[i].executDbSql(sql, self.db_name)
            self.log.info(result)
            self.assertNotIn('serial', result, '执行失败:' + step_txt)
        time.sleep(60 * 3 + 30)
        self.log.info('----创建操作同步，查询各节点数据库对象，备机有----')
        sql = f"select sysdate;select tgname from pg_trigger;"
        for i in range(2):
            result = self.sta_sh[i].executDbSql(sql, self.db_name)
            self.log.info(result)
            self.assertIn(self.trigger_name, result, '执行失败:' + step_txt)
        sql = f"select sysdate;\sf {self.func_name}"
        for i in range(2):
            result = self.sta_sh[i].executDbSql(sql, self.db_name)
            self.log.info(result)
            self.assertIn(self.func_name, result, '执行失败:' + step_txt)
        sql = f"select sysdate;\ds"
        for i in range(2):
            result = self.sta_sh[i].executDbSql(sql, self.db_name)
            self.log.info(result)
            self.assertIn('serial', result, '执行失败:' + step_txt)
        time.sleep(60*3)
        self.log.info('----删除操作同步，查询各节点数据库对象，备机无----')
        sql = f"select sysdate;select tgname from pg_trigger;"
        for i in range(2):
            result = self.sta_sh[i].executDbSql(sql, self.db_name)
            self.log.info(result)
            self.assertNotIn(self.trigger_name, result, '执行失败:' + step_txt)
        sql = f"select sysdate;\sf {self.func_name}"
        for i in range(2):
            result = self.sta_sh[i].executDbSql(sql, self.db_name)
            self.log.info(result)
            self.assertIn(self.constant.NOT_EXIST, result, '执行失败:' + step_txt)
        sql = f"select sysdate;\ds"
        for i in range(2):
            result = self.sta_sh[i].executDbSql(sql, self.db_name)
            self.log.info(result)
            self.assertNotIn('serial', result, '执行失败:' + step_txt)

    def tearDown(self):
        self.log.info('----环境清理----')
        self.log.info('----还原参数配置----')
        self.pri_sh.executeGsguc('reload',
                                 self.constant.GSGUC_SUCCESS_MSG,
                                 f'recovery_min_apply_delay=0')
        self.log.info('----还原集群----')
        self.sta_sh[0].buildStandby('-M standby')
        result = self.sta_sh[0].exec_refresh_conf()
        self.log.info(result)
        self.log.info('----重启数据库----')
        result = self.pri_sh.stopDbCluster()
        self.log.info(result)
        result = self.pri_sh.startDbCluster()
        self.log.info(result)
        self.log.info('----删除数据库对象----')
        drop_db_cmd = f'drop database {self.db_name};'
        result = self.pri_sh.executDbSql(drop_db_cmd)
        self.log.info(result)

        self.log.info('---Opengauss_Reliability_Dbsys_Case262.py 执行结束----')
